package com.gitee.kamismile.igniteEx.ignite;


import com.gitee.kamismile.igniteEx.common.IginteConfigProperties;
import com.gitee.kamismile.stone.commmon.util.ValueUtils;
import org.apache.ignite.*;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.failure.NoOpFailureHandler;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.logger.log4j2.Log4J2Logger;
import org.apache.ignite.services.Service;
import org.apache.ignite.services.ServiceConfiguration;
import org.apache.ignite.spi.discovery.DiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.event.EventListener;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static java.lang.Math.max;


@Component
public class IgniteServer {
    public static final int AVAILABLE_PROC_CNT = Runtime.getRuntime().availableProcessors();

    @Autowired
    private IginteConfigProperties iginteConfigProperties;

    private static final String LINE_NODE_PREFIX = "line.ignite.node.";

    @Autowired
    private ApplicationContext applicationContext;


    @Bean
    public String localHost() {
        return iginteConfigProperties.getClusterHost();
    }

    @Bean
    public DiscoverySpi discoverySpi() {
        TcpDiscoverySpi discoverySpi = new TcpDiscoverySpi();
        discoverySpi.setIpFinder(ipFinder());
        if(ValueUtils.isNotNull(iginteConfigProperties.getLocPort())){
            discoverySpi.setLocalPort(iginteConfigProperties.getLocPort());
        }
        //客户端重连 https://www.ignite-service.cn/doc/2.6.0/java/#_7-5-%E7%AE%A1%E7%90%86%E6%85%A2%E5%AE%A2%E6%88%B7%E7%AB%AF
        //当一个客户端发现它与一个集群断开时，会为自己赋予一个新的节点id然后试图与该服务端重新连接。注意：这会产生一个副作用，就是当客户端重建连接时本地ClusterNode的id属性会发生变化，这意味着，如果业务逻辑依赖于这个id，就会受到影响。 当客户端处于一个断开状态并且试图重建与集群的连接过程中时，Ignite API会抛出一个特定的异常：
        discoverySpi.setClientReconnectDisabled(true);
        return discoverySpi;
    }

    @Bean
    public TcpDiscoveryIpFinder ipFinder() {
        switch (ValueUtils.isIntegerNull(iginteConfigProperties.getIpFinderType(),1)){
            case 2:
                return getTcpDiscoveryMulticastIpFinder();
            default:
                return getTcpDiscoveryVmIpFinder();
        }
    }

    @NotNull
    private TcpDiscoveryMulticastIpFinder getTcpDiscoveryMulticastIpFinder() {
        TcpDiscoveryMulticastIpFinder ipFinder = new TcpDiscoveryMulticastIpFinder();
        ipFinder.setAddresses(addresses());
        return ipFinder;
    }

    @NotNull
    private TcpDiscoveryVmIpFinder getTcpDiscoveryVmIpFinder() {
        TcpDiscoveryVmIpFinder ipFinder = new TcpDiscoveryVmIpFinder();
        ipFinder.setAddresses(addresses());
        return ipFinder;
    }
    @Bean
    public List<String> addresses() {
        return iginteConfigProperties.getMembers();
    }

    @Bean
    public IgniteConfiguration igniteConfiguration() throws IgniteCheckedException, IOException {
        String appName = iginteConfigProperties.getUserAttributes().get("appName");
        String appVersion = iginteConfigProperties.getUserAttributes().get("appVersion");

        IgniteConfiguration cfg = new IgniteConfiguration();
        cfg.setClientMode(false);
        if(ValueUtils.isNotNull(iginteConfigProperties.getClusterHost())){
            cfg.setLocalHost(localHost());
        }
        cfg.setDiscoverySpi(discoverySpi());
        cfg.setConsistentId(iginteConfigProperties.getRoom() + "_" + appName + "_" + appVersion);
        cfg.setIgniteInstanceName(LINE_NODE_PREFIX + iginteConfigProperties.getRoom() + "_" + appName + "_" + appVersion);
        if (!iginteConfigProperties.getUserAttributes().isEmpty()) {
            cfg.setUserAttributes(iginteConfigProperties.getUserAttributes());
        }
        cfg.setMetricsLogFrequency(0);
        cfg.setNetworkTimeout(ValueUtils.isLongNull(iginteConfigProperties.getNetTimeout(),IgniteConfiguration.DFLT_NETWORK_TIMEOUT));
        cfg.setSystemWorkerBlockedTimeout(ValueUtils.isLongNull(iginteConfigProperties.getSysWorkerBlockedTimeout(),IgniteConfiguration.DFLT_FAILURE_DETECTION_TIMEOUT));
        cfg.setFailureHandler(new NoOpFailureHandler());
        cfg.setFailureDetectionTimeout(ValueUtils.isLongNull(iginteConfigProperties.getFailureDetectionTimeout(),IgniteConfiguration.DFLT_FAILURE_DETECTION_TIMEOUT));
        cfg.setGridLogger(new Log4J2Logger(new ClassPathResource("./log4j2.xml").getURL()));
        Map<String, Service> map = applicationContext.getBeansOfType(Service.class);

        ServiceConfiguration[] serviceConfigurations = map.keySet().parallelStream().map(v -> {
            ServiceConfiguration configuration = new ServiceConfiguration();
            configuration.setService(map.get(v));
            configuration.setName(v + "_" + appName + "_" + appVersion);
            configuration.setMaxPerNodeCount(iginteConfigProperties.getMaxPerNodeCount());
            configuration.setNodeFilter(new IgnitePredicate<ClusterNode>() {
                @Override
                public boolean apply(ClusterNode clusterNode) {
                    return clusterNode.attributes().containsValue(appName) &&
                            clusterNode.attributes().containsValue(appVersion);
                }
            });

            return configuration;
        }).collect(Collectors.toList()).toArray(new ServiceConfiguration[map.keySet().size()]);
//        cfg.setPeerClassLoadingEnabled(true);
        cfg.setServiceConfiguration(serviceConfigurations);
        cfg.setServiceThreadPoolSize(ValueUtils.isIntegerNull(iginteConfigProperties.getServiceThreadPoolSize(),max(8, AVAILABLE_PROC_CNT)));
        cfg.setSystemThreadPoolSize(ValueUtils.isIntegerNull(iginteConfigProperties.getSystemThreadPoolSize(),max(8, AVAILABLE_PROC_CNT)));
        cfg.setPublicThreadPoolSize(ValueUtils.isIntegerNull(iginteConfigProperties.getPublicThreadPoolSize(),max(8, AVAILABLE_PROC_CNT)));
        cfg.setAsyncCallbackPoolSize(ValueUtils.isIntegerNull(iginteConfigProperties.getAsyncCallbackPoolSize(),max(8, AVAILABLE_PROC_CNT)));
        cfg.setUtilityCachePoolSize(ValueUtils.isIntegerNull(iginteConfigProperties.getUtilityCachePoolSize(),max(8, AVAILABLE_PROC_CNT)));
        return cfg;
    }


    @Bean
    public IgniteSpringBean igniteSpringBean(IgniteConfiguration configuration) {
        IgniteSpringBean igniteSpringBean = new IgniteSpringBean();
        igniteSpringBean.setConfiguration(configuration);
        return igniteSpringBean;
    }


//    @Bean
//    public Ignite ignite(IgniteConfiguration igniteConfig, ApplicationContext context) throws IgniteCheckedException {
//        return IgniteSpring.start(igniteConfig, context);
//    }

    @EventListener
    public void shutdownHook(ApplicationReadyEvent event) {
        String appName = iginteConfigProperties.getUserAttributes().get("appName");
        String appVersion = iginteConfigProperties.getUserAttributes().get("appVersion");

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            Ignition.stop(LINE_NODE_PREFIX + iginteConfigProperties.getRoom() + "_" + appName + "_" + appVersion,false);
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }));
    }

}
